package org.apache.rocketmq.proxy.service.relay;

import io.netty.channel.*;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.service.channel.SimpleChannel;
import org.apache.rocketmq.proxy.service.transaction.TransactionData;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ConsumeMessageDirectlyResultRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;

import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.concurrent.CompletableFuture;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/13
 * @方法描述：GrpcClientChannel和RemotingChannel的公共父类
 */
public abstract class ProxyChannel extends SimpleChannel {

    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
    //表示远程Socket地址，也就是和proxy建立连接的客户端地址
    protected final SocketAddress remoteSocketAddress;
    //proxy本地Socket地址
    protected final SocketAddress localSocketAddress;
    //这个成员变量很重要，在GrpcClientChannel类的注释中大家肯定看到了，Broker节点会通过proxy向客户端主动发送消息
    //proxy接收到客户端的响应后会把响应返回给Broker节点，那么在中间起到代理转发响应的就是这个成员变量
    //也就是proxy接收到的客户端响应会通过这个成员变量转发给Broker节点
    protected final ProxyRelayService proxyRelayService;

    //构造方法
    protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, String remoteAddress, String localAddress) {
        super(parent, remoteAddress, localAddress);
        this.proxyRelayService = proxyRelayService;
        this.remoteSocketAddress = NetworkUtil.string2SocketAddress(remoteAddress);
        this.localSocketAddress = NetworkUtil.string2SocketAddress(localAddress);
    }

    //构造方法
    protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress, String localAddress) {
        super(parent, id, remoteAddress, localAddress);
        this.proxyRelayService = proxyRelayService;
        this.remoteSocketAddress = NetworkUtil.string2SocketAddress(remoteAddress);
        this.localSocketAddress = NetworkUtil.string2SocketAddress(localAddress);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：现在我可以跟大家进一步解释，如果使用本地模式部署了Proxy，这也就意味着proxy和Broker节点部署在同一个进程中，那么当broker节点想和客户端进行通信时，就比如说
     * 想获得消费者客户端的运行信息，就会调用下面这个方法，当然，我现在这么说大家肯定还不知道是怎么回事，还是那句话，别着急，等看到客户端信息注册到生产者管理器和消费者管理器时就清楚这是怎么回事了
     * 现在大家只需要记住这个方法会被Broker节点调用即可
     */
    @Override
    public ChannelFuture writeAndFlush(Object msg) {
        CompletableFuture<Void> processFuture = new CompletableFuture<>();

        try {//因为Broker节点没有使用grpc框架，所以Broker节点发送的肯定是RemotingCommand消息
            if (msg instanceof RemotingCommand) {
                //所以如果程序进入到这个分支，就意味着是Broker节点想和服务端进行通信，然后把消息发送给了proxy，接下来proxy要把消息发送给客户端
                //这里得到Proxy上下文对象
                ProxyContext context = ProxyContext.createForInner(this.getClass())
                        .setRemoteAddress(remoteAddress)
                        .setLocalAddress(localAddress);
                RemotingCommand command = (RemotingCommand) msg;
                if (command.getExtFields() == null) {
                    command.setExtFields(new HashMap<>());
                }
                //接下来就是要判断Broker想要对客户端执行什么操做了
                switch (command.getCode()) {
                    //这个是和事务相关的操作，可以先不必关注
                    case RequestCode.CHECK_TRANSACTION_STATE: {
                        CheckTransactionStateRequestHeader header = (CheckTransactionStateRequestHeader) command.readCustomHeader();
                        MessageExt messageExt = MessageDecoder.decode(ByteBuffer.wrap(command.getBody()), true, false, false);
                        RelayData<TransactionData, Void> relayData = this.proxyRelayService.processCheckTransactionState(context, command, header, messageExt);
                        processFuture = this.processCheckTransaction(header, messageExt, relayData.getProcessResult(), relayData.getRelayFuture());
                        break;
                    }//这个是想要获取消费者客户端运行信息的操作，接下来我就以这个方法为例子，讲解一下ProxyRelayService成员变量是如何发挥作用的
                    case RequestCode.GET_CONSUMER_RUNNING_INFO: {
                        //这里得到了Broker发送过来的请求的请求头
                        GetConsumerRunningInfoRequestHeader header = (GetConsumerRunningInfoRequestHeader) command.readCustomHeader();
                        //在这里调用了proxyRelayService成员变量的processGetConsumerRunningInfo方法，那现在请大家直接去LocalProxyRelayService类中查看相关注释
                        //看完了LocalProxyRelayService类的方法后，大家也知道了该类会返回一个future对象，也就是现在接收到的relayFuture对象
                        CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> relayFuture = this.proxyRelayService.processGetConsumerRunningInfo(context, command, header);
                        //在这里执行子类的方法，也就是GrpcClientChannel中的同名方法，而在执行子类方法中的时候，会把relayFuture放到channel管理器的resultNonceFutureMap表中
                        //当proxy接收到客户端回复的响应后，这个relayFuture就会被设置为完成状态，并且获得了客户端回复的响应内容，这一点可以在ClientActivity类中得到印证
                        //这样一来注册到relayFuture上的回调方法也就会执行了，在回调方法中就会把客户端回复的响应内容转发给Broker节点
                        processFuture = this.processGetConsumerRunningInfo(command, header, relayFuture);
                        break;
                    }//这里是想要消费者客户端直接消费消息的操作
                    case RequestCode.CONSUME_MESSAGE_DIRECTLY: {
                        ConsumeMessageDirectlyResultRequestHeader header = (ConsumeMessageDirectlyResultRequestHeader) command.readCustomHeader();
                        MessageExt messageExt = MessageDecoder.decode(ByteBuffer.wrap(command.getBody()), true, false, false);
                        processFuture = this.processConsumeMessageDirectly(command, header, messageExt, this.proxyRelayService.processConsumeMessageDirectly(context, command, header));
                        break;
                    }
                    default:
                        break;
                }
            } else {
                //如果进入到这个分支，就意味着要发送的消息不是RemotingCommand类型的
                //这也就意味着是proxy服务端自己想和客户端通信，注意，这里就和Broker节点没有关系了
                //是Proxy服务端自己想和客户端通信
                processFuture = processOtherMessage(msg);
            }
        } catch (Throwable t) {
            log.error("process failed. msg:{}", msg, t);
            processFuture.completeExceptionally(t);
        }
        DefaultChannelPromise promise = new DefaultChannelPromise(this, GlobalEventExecutor.INSTANCE);
        processFuture.thenAccept(ignore -> promise.setSuccess())
                .exceptionally(t -> {
                    promise.setFailure(t);
                    return null;
                });
        return promise;
    }

    //其他方法我就不添加注释了，有的在子类中已经讲过了，有的就是netty中的方法，都是空实现，也很简单，大家自己看看就行

    protected abstract CompletableFuture<Void> processOtherMessage(Object msg);

    protected abstract CompletableFuture<Void> processCheckTransaction(
            CheckTransactionStateRequestHeader header,
            MessageExt messageExt,
            TransactionData transactionData,
            CompletableFuture<ProxyRelayResult<Void>> responseFuture);

    protected abstract CompletableFuture<Void> processGetConsumerRunningInfo(
            RemotingCommand command,
            GetConsumerRunningInfoRequestHeader header,
            CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> responseFuture);

    protected abstract CompletableFuture<Void> processConsumeMessageDirectly(
            RemotingCommand command,
            ConsumeMessageDirectlyResultRequestHeader header,
            MessageExt messageExt,
            CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> responseFuture);

    @Override
    public ChannelConfig config() {
        return null;
    }

    @Override
    public ChannelMetadata metadata() {
        return null;
    }

    @Override
    protected AbstractUnsafe newUnsafe() {
        return null;
    }

    @Override
    protected boolean isCompatible(EventLoop loop) {
        return false;
    }

    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {

    }

    @Override
    protected void doDisconnect() throws Exception {

    }

    @Override
    protected void doClose() throws Exception {

    }

    @Override
    protected void doBeginRead() throws Exception {

    }

    @Override
    protected void doWrite(ChannelOutboundBuffer in) throws Exception {

    }

    @Override
    protected SocketAddress localAddress0() {
        return this.localSocketAddress;
    }

    @Override
    protected SocketAddress remoteAddress0() {
        return this.remoteSocketAddress;
    }
}
